package com.zhangdi.mserver.goes.core;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.zhangdi.mserver.goes.domain.ChangeInfo;
import com.zhangdi.mserver.goes.adapter.ParseAdapter;
import com.zhangdi.mserver.goes.config.MonitorOption;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by zhangdi on 2021/01/28 20:10:18.
 */

public class CanalMonitor implements Monitor {
    private Logger log = LoggerFactory.getLogger(CanalMonitor.class);
    private MonitorOption option;
    private CanalConnector connect;
    private Map<String, ParseAdapter> adapters;

    private volatile boolean running = false;
    private volatile Thread thread;

    public CanalMonitor() {
    }

    public CanalMonitor(MonitorOption option, CanalConnector connector) {
        this.option = option;
        this.connect = connector;
        this.adapters = new HashMap<>();
    }

    @Override
    public void start(List<ParseAdapter> adapters) {
        //TODO 重新监听 需要重新连接
        if (running) {
            log.info("is running. please execute stop().");
            return;
        }

        for (ParseAdapter adapter : adapters) {
            this.topic(adapter);
        }

        log.info("Monitor Running...");
        log.info("Monitor Option, host={}:{}, dest={}, filter={}, batchSize={}, idle={}",
                option.getHostname(), option.getPort(), option.getDestination(),
                option.getSubscribe(), option.getBatchSize(), option.getEmptyIdle());

        thread = new Thread(this::run);
        thread.start();
    }

    @Override
    public void topic(ParseAdapter adapter) {
        if (adapters.containsKey(adapter.tableName())) {
            return;
        }
        adapters.put(adapter.tableName(), adapter);
    }

    @Override
    public void stop() {
        thread.interrupt();
        log.info("Canal Monitor Client, shutdown.");
    }

    private void run() {
        running = true;
        connect.connect();
        connect.subscribe(option.getSubscribe());
        connect.rollback();
        while (true) {
            try {
                log.info("check new data...");
                Thread.sleep(option.getEmptyIdle());
            } catch (InterruptedException e) {
                disconnect();
                return;
            }

            Message message = connect.getWithoutAck(option.getBatchSize());
            long batchId = message.getId();
            int size = message.getEntries().size();
            if (batchId > 0 && size > 0) {//需要把batchId给过去
                log.info("Monitor, current batchId:{}", batchId);
                try {
                    parse(message.getEntries());
                } catch (Exception e) {
                    log.info("Monitor Exception, current batchId:{}, Message:{}", batchId, e.getMessage());
                    connect.rollback(batchId);
                }
            }
            connect.ack(batchId);
        }
    }

    private void disconnect() {
        //TODO 是否需要记录上一次的位置
        if (null != connect) {
            connect.disconnect();
            running = false;
            thread = null;
            adapters.clear();
        }
    }

    private void parse(List<Entry> entries) throws Exception {
        for (Entry entry : entries) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN ||
                    entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            if (!adapters.containsKey(entry.getHeader().getTableName())) {
                continue;
            }

            log.info("Parsing Table={}", entry.getHeader().getTableName());

            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());

            assert rowChange != null;
            EventType eventType = rowChange.getEventType();

            for (RowData rowData : rowChange.getRowDatasList()) {
                if (eventType != EventType.DELETE &&
                        eventType != EventType.INSERT &&
                        eventType != EventType.UPDATE) {
                    continue;
                }

                ChangeInfo changeInfo = new ChangeInfo();
                changeInfo.setOp(eventType);
                changeInfo.setTable(entry.getHeader().getTableName());
                changeInfo.setBeforeColumns(rowData.getBeforeColumnsList());
                changeInfo.setColumns(rowData.getAfterColumnsList());

                if (eventType == EventType.DELETE) {
                    changeInfo.setPk(getPk(changeInfo.getBeforeColumns()));
                } else {
                    changeInfo.setPk(getPk(changeInfo.getColumns()));
                }

                //TODO 失败记录 并且不能ack
                this.parse(changeInfo);
            }
        }
    }

    private String getPk(List<Column> columns) {
        for (Column column : columns) {
            if (column.getIsKey()) {
                return column.getValue();
            }
        }

        return null;
    }

    private boolean parse(ChangeInfo changeInfo) {
        return adapters.get(changeInfo.getTable()).parse(changeInfo);
    }

    private void printColumn(List<Column> columns) {
        for (Column column : columns) {
            log.info("{} : old={}  new={}", column.getName(), column.getValue(), column.getUpdated());
        }
    }
}
